package com.jiaying.sync;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.bson.Document;

import com.jiaying.sync.config.MongoConfig; // 更新导入语句
import com.jiaying.sync.config.SqlServerConfig; // 更新导入语句
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoDatabase;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

/**
 * 使用Debezium同步MongoDB数据到SQL Server
 * 实现MongoDB变更数据捕获(CDC)并同步到SQL Server数据库
 */
public class DebeziumMongoToSqlSync {
    
    private DebeziumEngine<ChangeEvent<String, String>> engine;
    private ExecutorService executor;
    private HikariDataSource dataSource; // 使用HikariCP连接池
    private MongoConfig mongoConfig; // 更新类型引用
    private SqlServerConfig sqlServerConfig; // 添加SQL Server配置
    private Properties debeziumProperties;
    private Set<String> createdTables; // 缓存已创建的表名
    
    /**
     * 构造函数 - 初始化Debezium MongoDB连接器配置
     */
    public DebeziumMongoToSqlSync() {
        // 初始化MongoDB配置
        this.mongoConfig = new MongoConfig(); // 更新构造函数调用
        
        // 初始化SQL Server配置
        this.sqlServerConfig = new SqlServerConfig(); // 更新构造函数调用
        
        // 初始化已创建表的缓存
        this.createdTables = new HashSet<>();
        
        // 初始化Debezium配置属性
        initializeDebeziumProperties();
        
        // 初始化SQL Server连接池
        initializeSqlConnectionPool();
    }
    
    /**
     * 初始化Debezium配置属性
     */
    private void initializeDebeziumProperties() {
        debeziumProperties = new Properties();
        
        // 连接器基本配置
        debeziumProperties.setProperty("name", "mongo-connector");
        debeziumProperties.setProperty("connector.class", "io.debezium.connector.mongodb.MongoDbConnector");
        
        // 使用连接字符串并指定副本集名称（根据错误信息添加）
        String connectionString = "mongodb://" + mongoConfig.getClusterHosts() + "/?replicaSet=rs0";
        debeziumProperties.setProperty("mongodb.connection.string", connectionString);
        
        // 添加topic前缀（Debezium新版本要求）
        debeziumProperties.setProperty("topic.prefix", "mongo-cluster-test");
        
        debeziumProperties.setProperty("mongodb.name", "mongo-server");
        debeziumProperties.setProperty("database.include.list", mongoConfig.getDatabaseName());
        
        // 使用通配符监听多个表/集合
        // 格式: databaseName.collectionPattern
        // 例如: testdb.* 会监听testdb数据库中的所有集合
        // 例如: testdb.coll.* 会监听testdb数据库中以coll开头的所有集合
        // 例如: testdb.(collection1|collection2) 会监听testdb数据库中的collection1和collection2集合
        String collectionPattern = System.getProperty("debezium.collection.pattern", "*");
        debeziumProperties.setProperty("collection.include.list", 
            mongoConfig.getDatabaseName() + "." + collectionPattern);
        
        // 偏移量存储配置（不使用Kafka）
        debeziumProperties.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        debeziumProperties.setProperty("offset.storage.file.filename", "offsets.dat");
        debeziumProperties.setProperty("offset.flush.interval.ms", "60000");
        
        // 心跳配置
        debeziumProperties.setProperty("heartbeat.interval.ms", "10000");
        
        // 快照配置
        debeziumProperties.setProperty("snapshot.mode", "initial");
        
        // 高级配置：监听新增的集合
        // 当设置为true时，连接器会定期扫描新的集合并自动开始监听
        debeziumProperties.setProperty("collection.include.list.change.enabled", "true");
        // 扫描间隔（毫秒）
        debeziumProperties.setProperty("collection.include.list.change.interval.ms", "30000");
        
        System.out.println("Debezium配置属性初始化完成");
        System.out.println("MongoDB连接字符串: " + connectionString);
        System.out.println("监听配置: database=" + mongoConfig.getDatabaseName() + ", collection pattern=" + collectionPattern);
    }
    
    /**
     * 初始化SQL Server连接池
     */
    private void initializeSqlConnectionPool() {
        try {
            // 创建HikariCP配置
            HikariConfig config = new HikariConfig();
            config.setJdbcUrl("jdbc:sqlserver://" + sqlServerConfig.getHost() + ":" + sqlServerConfig.getPort() + ";databaseName=" + sqlServerConfig.getDatabaseName());
            config.setUsername(sqlServerConfig.getUsername());
            config.setPassword(sqlServerConfig.getPassword());
            
            // 连接池配置（从配置文件读取）
            config.setMaximumPoolSize(sqlServerConfig.getMaxPoolSize());
            config.setMinimumIdle(sqlServerConfig.getMinIdle());
            config.setConnectionTimeout(sqlServerConfig.getConnectionTimeoutMs());
            config.setIdleTimeout(sqlServerConfig.getIdleTimeoutMs());
            config.setMaxLifetime(sqlServerConfig.getMaxLifetimeMs());
            config.setLeakDetectionThreshold(sqlServerConfig.getLeakDetectionThresholdMs());
            
            // 设置连接池名称
            config.setPoolName("DebeziumSyncPool");
            
            // 创建Hikari数据源
            this.dataSource = new HikariDataSource(config);
            
            System.out.println("SQL Server连接池初始化成功");
        } catch (Exception e) {
            System.err.println("SQL Server连接池初始化失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 从连接池获取连接
     * @return SQL Server连接
     * @throws SQLException
     */
    private Connection getConnection() throws SQLException {
        if (dataSource != null) {
            return dataSource.getConnection();
        } else {
            throw new SQLException("连接池未初始化");
        }
    }
    
    /**
     * 初始化SQL Server连接（保持向后兼容）
     */
    private void initializeSqlConnection() {
        // 此方法现在是空的，因为连接池已在initializeSqlConnectionPool中初始化
        // 保留此方法以确保代码兼容性
    }
    
    /**
     * 创建目标表（如果不存在）
     */
    private void createTargetTable() {
        Connection conn = null;
        PreparedStatement statement = null;
        try {
            conn = getConnection(); // 从连接池获取连接
            String createTableSQL = "IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='MongoData' AND xtype='U') " +
                                   "CREATE TABLE MongoData (" +
                                   "id NVARCHAR(255) PRIMARY KEY, " +
                                   "data NVARCHAR(MAX), " +
                                   "operation NVARCHAR(10), " +
                                   "timestamp DATETIME DEFAULT GETDATE()" +
                                   ")";
            
            statement = conn.prepareStatement(createTableSQL);
            statement.execute();
            
            System.out.println("目标表创建完成");
        } catch (SQLException e) {
            System.err.println("创建目标表失败: " + e.getMessage());
        } finally {
            // 关闭资源
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    System.err.println("关闭PreparedStatement失败: " + e.getMessage());
                }
            }
            if (conn != null) {
                try {
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    System.err.println("关闭Connection失败: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 启动Debezium引擎
     * 开始监听MongoDB变更事件并同步到SQL Server
     */
    public void startSync() {
        try {
            // 创建Debezium引擎
            engine = DebeziumEngine.create(Json.class)
                .using(debeziumProperties)
                .notifying(this::handleChangeEvent)
                .build();
            
            // 创建线程池并启动引擎
            executor = Executors.newSingleThreadExecutor();
            executor.execute(engine);
            
            System.out.println("Debezium同步引擎已启动");
            System.out.println("开始监听MongoDB变更事件并同步到SQL Server...");
            
        } catch (Exception e) {
            System.err.println("启动Debezium引擎失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 处理变更事件
     * @param event 变更事件
     */
    private void handleChangeEvent(ChangeEvent<String, String> event) {
        try {
            // 解析变更事件
            String key = event.key();
            String value = event.value();
            
            // System.out.println("接收到变更事件:");
            // System.out.println("Key: " + key);
            // System.out.println("Value: " + value);
            
            // 将变更同步到SQL Server
            syncToSqlServer(key, value);
            
        } catch (Exception e) {
            System.err.println("处理变更事件失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 同步数据到SQL Server
     * @param key 事件键
     * @param value 事件值
     */
    private void syncToSqlServer(String key, String value) {
        Connection conn = null;
        try {
            // 从连接池获取连接
            conn = getConnection();
            
            // 解析MongoDB变更事件
            Document eventDoc = Document.parse(value);
            Document payload = eventDoc.get("payload", new Document());
            
            String operation = payload.getString("op"); // 操作类型：r(读取快照), c(创建), u(更新), d(删除)
            
            // 获取源信息，包括数据库和集合名称
            Document source = payload.get("source", new Document());
            String databaseName = source.getString("db");
            String collectionName = source.getString("collection");
            
            // 构造SQL表名（可以基于MongoDB集合名）
            String sqlTableName = "MongoData_" + collectionName; // 或者使用其他命名规则
            
            Document after = null;
            
            // 安全地获取after字段
            Object afterObj = payload.get("after");
            if (afterObj != null) {
                if (afterObj instanceof Document) {
                    after = (Document) afterObj;
                } else if (afterObj instanceof String) {
                    // 如果after是字符串，则解析为Document
                    try {
                        after = Document.parse((String) afterObj);
                    } catch (Exception e) {
                        System.err.println("解析after字段为Document时出错: " + e.getMessage());
                        after = new Document();
                    }
                } else {
                    // 其他情况，创建空的Document
                    after = new Document();
                }
            } else {
                after = new Document();
            }
            
            // 对于快照事件（op="r"），可能没有_id字段在after中，需要从key中获取
            String id = null;
            if (after != null && !after.isEmpty()) {
                // 尝试从after文档中获取_id
                Object idObj = after.get("_id");
                if (idObj != null) {
                    id = extractIdFromObject(idObj);
                }
            }
            
            // 如果after中没有_id，则尝试从key中解析
            if (id == null || id.isEmpty()) {
                try {
                    // 解析key来获取_id
                    Document keyDoc = Document.parse(key);
                    Document keyPayload = keyDoc.get("payload", new Document());
                    Object keyIdObj = keyPayload.get("id");
                    if (keyIdObj != null) {
                        id = extractIdFromObject(keyIdObj);
                    }
                } catch (Exception e) {
                    System.err.println("解析key时出错: " + e.getMessage());
                }
            }
            
            // 如果仍然没有获取到id，则记录警告并跳过
            if (id == null || id.isEmpty()) {
                System.err.println("无法从事件中提取文档ID，跳过此事件");
                System.out.println("Key: " + key);
                System.out.println("Value: " + value);
                return;
            }
            
            // 确保目标表存在
            createTargetTableIfNotExists(sqlTableName);
            
            // 准备SQL语句
            String sql = "";
            if ("r".equals(operation) || "c".equals(operation) || "u".equals(operation)) {
                // 读取快照、插入或更新操作
                sql = "MERGE " + sqlTableName + " AS target " +
                      "USING (SELECT ? AS id, ? AS data, ? AS operation, ? AS collection_name) AS source " +
                      "ON target.id = source.id " +
                      "WHEN MATCHED THEN " +
                      "UPDATE SET data = source.data, operation = source.operation, collection_name = source.collection_name, timestamp = GETDATE() " +
                      "WHEN NOT MATCHED THEN " +
                      "INSERT (id, data, operation, collection_name) VALUES (source.id, source.data, source.operation, source.collection_name);";
                
                PreparedStatement statement = conn.prepareStatement(sql);
                statement.setString(1, id);
                statement.setString(2, after != null ? after.toJson() : "{}");
                statement.setString(3, operation);
                statement.setString(4, collectionName); // 添加集合名称
                statement.executeUpdate();
                statement.close();
                
            } else if ("d".equals(operation)) {
                // 删除操作
                sql = "DELETE FROM " + sqlTableName + " WHERE id = ?";
                PreparedStatement statement = conn.prepareStatement(sql);
                statement.setString(1, id);
                statement.executeUpdate();
                statement.close();
            }
            
            System.out.println("数据同步到SQL Server成功: " + id + " (" + operation + ") from " + databaseName + "." + collectionName);
            
        } catch (SQLException e) {
            System.err.println("同步数据到SQL Server失败: " + e.getMessage());
            e.printStackTrace();
        } catch (Exception e) {
            System.err.println("解析变更事件失败: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 关闭连接，返回到连接池
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    System.err.println("关闭Connection失败: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 创建目标表（如果不存在），支持不同集合的表
     * @param tableName 表名
     */
    private synchronized void createTargetTableIfNotExists(String tableName) {
        // 检查表是否已经创建过
        if (createdTables.contains(tableName)) {
            return; // 表已经创建过，直接返回
        }
        
        Connection conn = null;
        PreparedStatement statement = null;
        try {
            conn = getConnection(); // 从连接池获取连接
            String createTableSQL = "IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=? AND xtype='U') " +
                                   "CREATE TABLE " + tableName + " (" +
                                   "id NVARCHAR(255) PRIMARY KEY, " +
                                   "data NVARCHAR(MAX), " +
                                   "operation NVARCHAR(10), " +
                                   "collection_name NVARCHAR(255), " +
                                   "timestamp DATETIME DEFAULT GETDATE()" +
                                   ")";
            
            statement = conn.prepareStatement(createTableSQL);
            statement.setString(1, tableName);
            statement.execute();
            
            // 将表名添加到已创建表的缓存中
            createdTables.add(tableName);
            
            System.out.println("目标表 " + tableName + " 创建完成或已存在");
        } catch (SQLException e) {
            System.err.println("创建目标表 " + tableName + " 失败: " + e.getMessage());
        } finally {
            // 关闭资源
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    System.err.println("关闭PreparedStatement失败: " + e.getMessage());
                }
            }
            if (conn != null) {
                try {
                    conn.close(); // 返回连接到池中
                } catch (SQLException e) {
                    System.err.println("关闭Connection失败: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 从对象中安全地提取ID
     * @param idObj ID对象
     * @return 提取的ID字符串
     */
    private String extractIdFromObject(Object idObj) {
        String id = null;
        if (idObj != null) {
            try {
                // 检查是否是Document类型（包含$oid的对象）
                if (idObj instanceof Document) {
                    Document idDoc = (Document) idObj;
                    Object oidObj = idDoc.get("$oid");
                    if (oidObj != null) {
                        id = oidObj.toString();
                    } else {
                        id = idDoc.toJson(); // 备用方案
                    }
                // } else if (idObj instanceof String) {
                //     String idStr = (String) idObj;
                //     // 检查是否是JSON格式的ObjectId
                //     if (idStr.startsWith("{") && idStr.contains("$oid")) {
                //         // 使用正则表达式安全地提取ObjectId
                //         id = extractObjectId(idStr);
                //     } else {
                //         id = idStr;
                //     }
                } else {
                    id = idObj.toString();
                }
            } catch (Exception e) {
                System.err.println("提取ID时出错: " + e.getMessage());
                id = null;
            }
        }
        return id;
    }
    
    /**
     * 从JSON字符串中安全地提取ObjectId
     * @param jsonStr 包含ObjectId的JSON字符串
     * @return 提取的ObjectId字符串
     */
    // private String extractObjectId(String jsonStr) {
    //     try {
    //         // 使用正则表达式提取ObjectId
    //         // 匹配格式: {"$oid": "507f1f77bcf86cd799439011"}
    //         java.util.regex.Pattern pattern = java.util.regex.Pattern.compile("\"\\$oid\"\\s*:\\s*\"([a-fA-F0-9]+)\"");
    //         java.util.regex.Matcher matcher = pattern.matcher(jsonStr);
    //         if (matcher.find()) {
    //             return matcher.group(1);
    //         }
    //     } catch (Exception e) {
    //         System.err.println("提取ObjectId时出错: " + e.getMessage());
    //     }
    //     // 如果无法提取，返回原始字符串
    //     return jsonStr;
    // }
    
    /**
     * 停止Debezium引擎
     * 停止监听并释放资源
     */
    public void stopSync() {
        try {
            // 停止Debezium引擎
            if (engine != null) {
                engine.close();
            }
            
            // 关闭线程池
            if (executor != null) {
                executor.shutdown();
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
            }
            
            // 关闭连接池
            if (dataSource != null) {
                dataSource.close();
            }
            
            System.out.println("Debezium同步引擎已停止");
            
        } catch (Exception e) {
            System.err.println("停止Debezium引擎失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 测试MongoDB连接
     */
    public void testMongoConnection() {
        try {
            String connectionString = "mongodb://" + mongoConfig.getClusterHosts();
            MongoClient mongoClient = MongoClients.create(connectionString);
            
            MongoDatabase database = mongoClient.getDatabase(mongoConfig.getDatabaseName());
            System.out.println("MongoDB连接测试成功");
            System.out.println("数据库名称: " + database.getName());
            
            mongoClient.close();
        } catch (Exception e) {
            System.err.println("MongoDB连接测试失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 测试SQL Server连接
     */
    public void testSqlConnection() {
        Connection conn = null;
        try {
            // 从连接池获取连接
            conn = getConnection();
            
            if (conn != null && !conn.isClosed()) {
                System.out.println("SQL Server连接测试成功");
                System.out.println("数据库: " + conn.getCatalog());
            } else {
                System.out.println("SQL Server连接已关闭");
            }
        } catch (SQLException e) {
            System.err.println("SQL Server连接测试失败: " + e.getMessage());
        } finally {
            // 关闭连接，返回到连接池
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    System.err.println("关闭Connection失败: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 主方法 - 演示Debezium MongoDB到SQL Server同步功能
     */
    public static void main(String[] args) {
        DebeziumMongoToSqlSync sync = new DebeziumMongoToSqlSync();
        
        // 添加关闭钩子以实现优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("\n接收到关闭信号，正在优雅关闭...");
            sync.stopSync();
            System.out.println("程序已优雅关闭");
        }));
        
        try {
            // 解析命令行参数
            parseArguments(args);
            
            // 测试连接
            System.out.println("=== 连接测试 ===");
            sync.testMongoConnection();
            sync.testSqlConnection();
            
            // 创建目标表
            sync.createTargetTable();
            
            // 启动同步
            System.out.println("\n=== 启动同步 ===");
            sync.startSync();
            
            System.out.println("同步引擎已启动，将持续运行...");
            System.out.println("按 Ctrl+C 或发送终止信号来优雅关闭程序");
            
            // 检查是否提供了运行时间参数
            long runTime = -1; // 默认持续运行
            boolean showStatus = false;
            
            for (int i = 0; i < args.length; i++) {
                if ("-t".equals(args[i]) && i + 1 < args.length) {
                    try {
                        runTime = Long.parseLong(args[i + 1]) * 1000; // 转换为毫秒
                        System.out.println("程序将在 " + args[i + 1] + " 秒后自动停止");
                        i++; // 跳过下一个参数
                    } catch (NumberFormatException e) {
                        System.err.println("无效的运行时间参数: " + args[i + 1]);
                    }
                } else if ("-s".equals(args[i])) {
                    showStatus = true;
                    System.out.println("将显示运行状态信息");
                }
            }
            
            if (runTime < 0) {
                System.out.println("程序将持续运行直到收到关闭信号");
            }
            
            long startTime = System.currentTimeMillis();
            long lastStatusTime = startTime;
            // 持续运行直到收到关闭信号或达到指定运行时间
            while (!Thread.currentThread().isInterrupted()) {
                // 如果指定了运行时间且已达到，则退出
                if (runTime > 0 && (System.currentTimeMillis() - startTime) >= runTime) {
                    System.out.println("达到指定运行时间，正在停止...");
                    break;
                }
                
                try {
                    // 每隔1秒检查一次状态
                    Thread.sleep(1000);
                    
                    // 如果需要显示状态信息
                    if (showStatus && (System.currentTimeMillis() - lastStatusTime) >= 30000) { // 每30秒显示一次
                        System.out.println("同步引擎正在运行中... 运行时间: " + 
                            (System.currentTimeMillis() - startTime) / 1000 + " 秒");
                        lastStatusTime = System.currentTimeMillis();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            
        } catch (Exception e) {
            System.err.println("同步过程中发生错误: " + e.getMessage());
            e.printStackTrace();
        } finally {
            System.out.println("正在关闭同步引擎...");
            sync.stopSync();
        }
    }
    
    /**
     * 解析命令行参数
     * @param args 命令行参数
     */
    private static void parseArguments(String[] args) {
        for (int i = 0; i < args.length; i++) {
            if ("-p".equals(args[i]) && i + 1 < args.length) {
                // 设置集合模式
                System.setProperty("debezium.collection.pattern", args[i + 1]);
                System.out.println("设置集合监听模式: " + args[i + 1]);
                i++; // 跳过下一个参数
            }
        }
    }
}